package per.chao.mqtt.application;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import per.chao.mqtt.domain.protocol.ProtocolTransportMap;

import java.io.IOException;

/**
 * Description:
 *
 * @author W.Chao
 * @date 2021/6/19 23:09
 **/
@ChannelHandler.Sharable
@Component
@Slf4j
public class ProtocolTransportHandler extends SimpleChannelInboundHandler<MqttMessage> {
	@Autowired
	private ProtocolTransportMap transportProtocols;

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, MqttMessage mqttMessage) throws Exception {
		if (mqttMessage.decoderResult().isFailure()) {
			Throwable cause = mqttMessage.decoderResult().cause();
			if (cause instanceof MqttUnacceptableProtocolVersionException) {
				ctx.writeAndFlush(MqttMessageFactory.newMessage(
						new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
						new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION, false),
						null));
				log.error("", cause);
			} else if (cause instanceof MqttIdentifierRejectedException) {
				ctx.writeAndFlush(MqttMessageFactory.newMessage(
						new MqttFixedHeader(MqttMessageType.CONNACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
						new MqttConnAckVariableHeader(MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED, false),
						null));
				log.error("", cause);
			} else {
				log.error("other error", cause);
				ctx.close();
				return;
			}
		}

		Channel channel = ctx.channel();

		log.info(String.valueOf(mqttMessage.fixedHeader().messageType()));
		switch (mqttMessage.fixedHeader().messageType()) {
			case PUBACK:
				transportProtocols.getProtocol(MqttMessageType.PUBACK).handleMessage(channel, mqttMessage.variableHeader());
				break;
			case PUBREC:
				transportProtocols.getProtocol(MqttMessageType.PUBREC).handleMessage(channel, mqttMessage.variableHeader());
				break;
			case PUBREL:
				transportProtocols.getProtocol(MqttMessageType.PUBREL).handleMessage(channel, mqttMessage.variableHeader());
				break;
			case CONNECT:
				transportProtocols.getProtocol(MqttMessageType.CONNECT).handleMessage(channel, mqttMessage);
				break;
			case PINGREQ:
				transportProtocols.getProtocol(MqttMessageType.PINGREQ).handleMessage(channel, mqttMessage);
				break;
			case PUBCOMP:
				transportProtocols.getProtocol(MqttMessageType.PUBCOMP).handleMessage(channel, mqttMessage.variableHeader());
				break;
			case PUBLISH:
				transportProtocols.getProtocol(MqttMessageType.PUBLISH).handleMessage(channel, mqttMessage);
				break;
			case SUBSCRIBE:
				transportProtocols.getProtocol(MqttMessageType.SUBSCRIBE).handleMessage(channel, mqttMessage);
				break;
			case DISCONNECT:
				transportProtocols.getProtocol(MqttMessageType.DISCONNECT).handleMessage(channel, mqttMessage);
				break;
			case UNSUBSCRIBE:
				transportProtocols.getProtocol(MqttMessageType.UNSUBSCRIBE).handleMessage(channel, mqttMessage);
				break;
			case UNSUBACK:
				transportProtocols.getProtocol(MqttMessageType.UNSUBACK).handleMessage(channel, mqttMessage.variableHeader());
				break;
			case PINGRESP:
			case CONNACK:
			case SUBACK:
			default:
				break;
		}
	}


	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		super.channelActive(ctx);
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		super.channelInactive(ctx);
	}

	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
		super.userEventTriggered(ctx, evt);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		if (cause instanceof IOException) {
			ctx.close();
		} else {
			cause.getStackTrace();
			super.exceptionCaught(ctx, cause);
		}
	}
}
